package com.swak.job.storage;

import java.sql.Connection;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.sql.DataSource;

import org.springframework.beans.factory.InitializingBean;

import com.swak.App;
import com.swak.db.DbStorage;
import com.swak.job.Job;
import com.swak.job.JobDispatch;
import com.swak.job.JobTask;
import com.swak.job.enums.Status;
import com.swak.job.timer.TimerTask.TaskFrag;
import com.swak.utils.Converters;
import com.swak.utils.IOUtils;
import com.swak.utils.Ints;
import com.swak.utils.Lists;
import com.swak.utils.StringUtils;

public abstract class AbstractStorageService extends DbStorage implements StorageService, InitializingBean {

	protected String module;
	protected String jobTable;
	protected String jobDispatchTable;
	protected String jobTaskTable;
	protected String jobFailTable;

	public AbstractStorageService(DataSource dataSource, String module) {
		super(dataSource);
		this.module = module;
		this.jobTable = "JOB_" + this.module;
		this.jobDispatchTable = "JOB_" + this.module + "_DISPATCH";
		this.jobTaskTable = "JOB_" + this.module + "_TASK";
		this.jobFailTable = "JOB_" + this.module + "_FAIL";
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		String tables = IOUtils.readAsText(this.getClass(), "tables.sql");
		tables = StringUtils.replace(tables, "${jobTable}", this.jobTable);
		tables = StringUtils.replace(tables, "${jobDispatchTable}", this.jobDispatchTable);
		tables = StringUtils.replace(tables, "${jobTaskTable}", this.jobTaskTable);
		tables = StringUtils.replace(tables, "${jobFailTable}", this.jobFailTable);
		String[] jobs = StringUtils.split(tables, ";");
		for (String job : jobs) {
			this.executeUpdate(job);
		}
	}

	@Override
	public void create(Job job) {
		String sql = "INSERT INTO " + this.jobTable + "(ID, NAME, CRON_EXPRESSION, ORDER_FLAG)" + " VALUES(?,?,?,?)"
				+ "ON DUPLICATE KEY UPDATE NAME = ?, CRON_EXPRESSION = ?, ORDER_FLAG = ?";
		this.executeUpdate(sql, job.getId(), job.getName(), job.getCronExpression(), job.getOrder(), job.getName(),
				job.getCronExpression(), job.getOrder());
	}

	@Override
	public boolean dispatch(JobDispatch dispatch, TaskFrag frag) {
		AtomicBoolean canDispatch = new AtomicBoolean(false);
		this.executeMultiSql((conn) -> {
			Job myJob = this.lock(conn, dispatch.getJobId());
			if (myJob != null) {
				boolean dispatchAble = false;

				// 本次调度是否可以执行
				if (myJob.getNextDispatchTime() == null
						|| myJob.getNextDispatchTime().compareTo(frag.getRunTime()) <= 1) {
					dispatchAble = true;
				}

				// 如果需要顺序执行则需要等上一次调度全部执行结束后才能再次发起调度 -- 暂不支持
				if (dispatchAble && myJob.getOrder() && myJob.getLastDispatchId() != null) {
					JobDispatch lastDispatch = this.getJobDispatch(conn, myJob.getLastDispatchId());
					if (lastDispatch != null && lastDispatch.getDispatchs() != null
							&& lastDispatch.getExecutes() != null
							&& lastDispatch.getDispatchs() > lastDispatch.getExecutes()) {
						dispatchAble = false;
					}
				}
				canDispatch.set(dispatchAble);
			}
			if (canDispatch.get()) {
				this.doDispatch(conn, dispatch, frag);
			}
		});
		return canDispatch.get();
	}

	private Job lock(Connection connection, String id) {
		String sql = "SELECT ID, LAST_DISPATCH_ID, LAST_DISPATCH_TIME, NEXT_DISPATCH_TIME, ORDER_FLAG FROM "
				+ this.jobTable + " WHERE ID = ? FOR UPDATE";
		List<Map<String, Object>> datas = this.executeMultiSql_Query(connection, sql, id);
		if (datas != null && datas.size() == 1) {
			Map<String, Object> data = datas.get(0);
			Job job = new Job();
			job.setId(this.doConvert(data.get("ID"), String.class));
			job.setLastDispatchId(this.doConvert(data.get("LAST_DISPATCH_ID"), Long.class));
			job.setLastDispatchTime(this.doConvert(data.get("LAST_DISPATCH_TIME"), LocalDateTime.class));
			job.setNextDispatchTime(this.doConvert(data.get("NEXT_DISPATCH_TIME"), LocalDateTime.class));
			job.setOrder(this.doConvert(String.valueOf(data.get("ORDER_FLAG")), Boolean.class));
			return job;
		}
		return null;
	}

	private JobDispatch getJobDispatch(Connection connection, Long id) {
		String sql = "SELECT ID, URL, JOB_ID, JOB_NAME, CREATE_TIME, DISPATCHS, EXECUTES FROM " + this.jobDispatchTable
				+ " WHERE ID = ?";
		List<Map<String, Object>> datas = this.executeMultiSql_Query(connection, sql, id);
		if (datas != null && datas.size() == 1) {
			Map<String, Object> data = datas.get(0);
			JobDispatch dispatch = new JobDispatch();
			dispatch.setId(this.doConvert(data.get("ID"), Long.class));
			dispatch.setUrl(this.doConvert(data.get("URL"), String.class));
			dispatch.setJobId(this.doConvert(data.get("JOB_ID"), String.class));
			dispatch.setJobName(this.doConvert(data.get("JOB_NAME"), String.class));
			dispatch.setDispatchs(this.doConvert(data.get("DISPATCHS"), Integer.class));
			dispatch.setExecutes(this.doConvert(data.get("EXECUTES"), Integer.class));
			dispatch.setCreateTime(this.doConvert(data.get("CREATE_TIME"), LocalDateTime.class));
			return dispatch;
		}
		return null;
	}

	private void doDispatch(Connection connection, JobDispatch dispatch, TaskFrag frag) {
		dispatch.setId(App.genId());
		String dispatchSql = "INSERT INTO " + this.jobDispatchTable + "(ID, URL, JOB_ID, JOB_NAME, CREATE_TIME)"
				+ " VALUES(?,?,?,?,?)";
		this.executeMultiSql_Update(connection, dispatchSql, dispatch.getId(), dispatch.getUrl(), dispatch.getJobId(),
				dispatch.getJobName(), dispatch.getCreateTime());

		String updateSql = "UPDATE " + this.jobTable
				+ " SET LAST_DISPATCH_ID = ?, LAST_DISPATCH_TIME = ?, NEXT_DISPATCH_TIME = ? WHERE ID = ?";
		this.executeMultiSql_Update(connection, updateSql, dispatch.getId(), frag.getRunTime(), frag.getNextTime(),
				dispatch.getJobId());
	}

	@Override
	public void createTask(JobTask task) {
		task.setId(App.genId());
		String sql = "INSERT INTO " + this.jobTaskTable
				+ "(ID, EXECUTE_URL, DISPATCH_URL, DISPATCH_ID, JOB_ID, JOB_NAME, CREATE_TIME, LAST_EXECUTION_TIME, STATUS, CONFIG, RETRY_TIMES)"
				+ " VALUES(?,?,?,?,?,?,?,?,?,?,?)";
		this.executeUpdate(sql, task.getId(), task.getExecuteUrl(), task.getDispatchUrl(), task.getDispatchId(),
				task.getJobId(), task.getJobName(), task.getCreateTime(), task.getCreateTime(), task.getStatus(),
				task.getConfig(), task.getRetryTimes());
	}

	@Override
	public void dispatchTask(JobTask task) {
		this.executeMultiSql((conn) -> {
			String taskSql = "UPDATE " + this.jobTaskTable + " SET STATUS = ? WHERE ID = ?";
			this.executeMultiSql_Update(conn, taskSql, task.getStatus(), task.getId());
			String dispatchSql = "UPDATE " + this.jobDispatchTable + " SET DISPATCHS = DISPATCHS + 1 WHERE ID = ?";
			this.executeMultiSql_Update(conn, dispatchSql, task.getDispatchId());
		});
	}

	@Override
	public void execute(JobTask task) {
		if (task.getStatus() == Status.Success.getValue()) {
			this.executeMultiSql((conn) -> {
				// @TODO 数据不删除 -- 修改为执行成功
				String taskSql = "UPDATE " + this.jobTaskTable
						+ " SET STATUS = ?, RESULT = ?, LAST_EXECUTION_TIME =? WHERE ID = ?";
				this.executeMultiSql_Update(conn, taskSql, task.getStatus(), task.getResult(),
						task.getLastExecutionTime(), task.getId());
				String dispatchSql = "UPDATE " + this.jobDispatchTable + " SET EXECUTES = EXECUTES + 1 WHERE ID = ?";
				this.executeMultiSql_Update(conn, dispatchSql, task.getDispatchId());
			});
		} else if (task.getRetryTimes() <= 0) {
			this.executeMultiSql((conn) -> {
				String failSql = "INSERT INTO " + this.jobFailTable
						+ "(ID, EXECUTE_URL, DISPATCH_URL, DISPATCH_ID, JOB_ID, JOB_NAME, CREATE_TIME, LAST_EXECUTION_TIME, STATUS, CONFIG)"
						+ " VALUES(?,?,?,?,?,?,?,?,?,?)";
				this.executeMultiSql_Update(conn, failSql, task.getId(), task.getExecuteUrl(), task.getDispatchUrl(),
						task.getDispatchId(), task.getJobId(), task.getJobName(), task.getCreateTime(),
						task.getLastExecutionTime(), task.getStatus(), task.getConfig());

				// @TODO 数据不删除
				String taskSql = "UPDATE " + this.jobTaskTable
						+ " SET STATUS = ?, RESULT = ?, LAST_EXECUTION_TIME =? WHERE ID = ?";
				this.executeMultiSql_Update(conn, taskSql, task.getStatus(), task.getResult(),
						task.getLastExecutionTime(), task.getId());
			});
		} else {
			String sql = "UPDATE " + this.jobTaskTable
					+ " SET STATUS = ?, RESULT = ?, LAST_EXECUTION_TIME =?, RETRY_TIMES =? WHERE ID = ?";
			task.setRetryTimes(Ints.addI(task.getRetryTimes(), -1));
			this.executeUpdate(sql, task.getStatus(), task.getResult(), task.getLastExecutionTime(),
					task.getRetryTimes(), task.getId());
		}
	}

	/**
	 * 查询执行失败的需要重试的任务： 发送失败、执行失败
	 */
	@Override
	public List<JobTask> queryFailTasks(Date delayDate) {
		String sql = "SELECT ID, EXECUTE_URL, DISPATCH_URL, DISPATCH_ID, JOB_ID, JOB_NAME, LAST_EXECUTION_TIME, CREATE_TIME, CONFIG, STATUS, RETRY_TIMES FROM "
				+ this.jobTaskTable + " WHERE STATUS IN(-1,0) AND LAST_EXECUTION_TIME <= ? FOR UPDATE";
		List<Map<String, Object>> datas = this.executeQuery(sql, delayDate);
		List<JobTask> tasks = Lists.newArrayList();
		if (datas != null && datas.size() > 0) {
			for (Map<String, Object> data : datas) {
				JobTask task = new JobTask();
				task.setId(this.doConvert(data.get("ID"), Long.class));
				task.setExecuteUrl(this.doConvert(data.get("EXECUTE_URL"), String.class));
				task.setDispatchUrl(this.doConvert(data.get("DISPATCH_URL"), String.class));
				task.setDispatchId(this.doConvert(data.get("DISPATCH_ID"), Long.class));
				task.setJobId(this.doConvert(data.get("JOB_ID"), String.class));
				task.setJobName(this.doConvert(data.get("JOB_NAME"), String.class));
				task.setLastExecutionTime(this.doConvert(data.get("LAST_EXECUTION_TIME"), LocalDateTime.class));
				task.setCreateTime(this.doConvert(data.get("CREATE_TIME"), LocalDateTime.class));
				task.setConfig(this.doConvert(data.get("CONFIG"), String.class));
				task.setStatus(this.doConvert(data.get("STATUS"), Byte.class));
				task.setRetryTimes(this.doConvert(data.get("RETRY_TIMES"), Integer.class));
				tasks.add(task);
			}
		}
		return tasks;
	}

	@SuppressWarnings("unchecked")
	protected <T> T doConvert(Object value, Class<?> targetType) {
		return (T) Converters.doConvert(value, targetType);
	}
}
